package com.chief.lesson1;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class WordCount {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        String path="E:\\bigdata-start\\flink\\flinkCode\\src\\main\\java\\com\\chief\\lesson1\\hello.txt";

        DataSource<String> dataSource = env.readTextFile(path);
        AggregateOperator<Tuple2<String, Integer>> sum = dataSource.flatMap(new MyFlatMapFunction()).groupBy(0).sum(1);

        sum.print();

//        for (Tuple2<String, Integer> stringIntegerTuple2 : sum.collect()) {
//            System.out.println(stringIntegerTuple2.toString());
//        }
    }

    public static class MyFlatMapFunction implements FlatMapFunction<String, Tuple2<String ,Integer >> {

        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
            String[] s1 = s.split(" ");
            for (String s2 : s1) {
                collector.collect(new Tuple2<>(s2,1));
            }
        }
    }
}
